package com.atguigu.gmall.realtime.util;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.annotation.NotSink;
import com.atguigu.gmall.realtime.bean.TableProcess;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.sink.PhoenixSink;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.shaded.guava18.com.google.common.base.CaseFormat;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.lang.reflect.Field;
import java.nio.charset.StandardCharsets;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.Properties;
import java.util.stream.Collectors;

/**
 * @Author lzc
 * @Date 2022/12/27 15:05
 */
public class FlinkSinkUtil {
    
    public static SinkFunction<Tuple2<JSONObject, TableProcess>> getPhoenixSink() {
        return new PhoenixSink();
    }
    
    public static SinkFunction<String> getKafkaSink(String topic) {
        Properties props = new Properties();
        props.put("bootstrap.servers", Constant.KAFKA_BROKERS);
        
        // 这个参数不允许超过 kafka 服务器端的设置(15 分钟)(transaction.max.timeout.ms)
        props.put("transaction.timeout.ms", 15 * 60 * 1000);
        
        return new FlinkKafkaProducer<String>(
            "default",
            new KafkaSerializationSchema<String>() {
                @Override
                public ProducerRecord<byte[], byte[]> serialize(String element,
                                                                @Nullable Long timestamp) {
                    return new ProducerRecord<>(topic, element.getBytes(StandardCharsets.UTF_8));
                }
                
            },
            props,
            FlinkKafkaProducer.Semantic.EXACTLY_ONCE
        );
    }
    
    public static SinkFunction<Tuple2<JSONObject, TableProcess>> getKafkaSink() {
        Properties props = new Properties();
        props.put("bootstrap.servers", Constant.KAFKA_BROKERS);
        
        // 这个参数不允许超过 kafka 服务器端的设置(15 分钟)(transaction.max.timeout.ms)
        props.put("transaction.timeout.ms", 15 * 60 * 1000);
        
        return new FlinkKafkaProducer<Tuple2<JSONObject, TableProcess>>(
            "default",
            new KafkaSerializationSchema<Tuple2<JSONObject, TableProcess>>() {
                @Override
                public ProducerRecord<byte[], byte[]> serialize(Tuple2<JSONObject, TableProcess> t,
                                                                @Nullable Long timestamp) {
                    JSONObject data = t.f0;  // 要写出的数据
                    String topic = t.f1.getSinkTable();  // topic 的名字
                    return new ProducerRecord<>(topic, data.toJSONString().getBytes(StandardCharsets.UTF_8));
                }
            },
            props,
            FlinkKafkaProducer.Semantic.EXACTLY_ONCE
        );
    }
    
    public static <T> SinkFunction<T> getClickHouseSink(String table,
                                                        Class<T> tClass,
                                                        boolean ... isCamelToUnderline) {
        boolean flag = true; // 默认把驼峰转成下划线
        if (isCamelToUnderline.length > 0) {
            flag = isCamelToUnderline[0];
        }
        final boolean finalFlat = flag;
        
        // 直接使用 jdbc sink
        String driver = Constant.CLICKHOUSE_DRIVER;
        String url = Constant.CLICKHOUSE_URL;
        // insert into user(a,b,c)values(?,?,?)
        Field[] fields = tClass.getDeclaredFields();
        String columns = Arrays.stream(fields)
            .filter(field -> field.getAnnotation(NotSink.class) == null)
            .map(f -> {
                if (finalFlat) {
                    return CaseFormat.LOWER_CAMEL.to(CaseFormat.LOWER_UNDERSCORE, f.getName());
                }
                return f.getName();
            })
            .collect(Collectors.joining(","));
        StringBuilder sql = new StringBuilder();
        sql
            .append("insert into ")
            .append(table)
            .append("(")
            .append(columns) // 拼接字段名
            .append(")values(")
            .append(columns.replaceAll("[^,]+", "?")) // 拼接占位符
            .append(")");
        
        
        System.out.println("clickhouse 的插入语句:" + sql);
        return JdbcSink.sink(sql.toString(),
                             new JdbcStatementBuilder<T>() {
                                 @Override
                                 public void accept(PreparedStatement ps,
                                                    T t) throws SQLException {
                                     Field[] fields = tClass.getDeclaredFields();
                                     try {
                                         for (int i = 0, position = 1; i < fields.length; i++) {
                                             Field field = fields[i];
                                             if (field.getAnnotation(NotSink.class) == null) {
                                                 field.setAccessible(true);
                                                 Object v = field.get(t);
                                                 ps.setObject(position++, v);
                                             }
                                         }
                                     } catch (IllegalAccessException e) {
                                         throw new RuntimeException(e);
                                     }
                                 }
                             },
                             new JdbcExecutionOptions.Builder()
                                 .withBatchSize(1024 * 1024)  // 批次大小
                                 .withBatchIntervalMs(2000) // 批次的写出的时间间隔
                                 .withMaxRetries(3) // 重试次数
                                 .build(),
                             new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                                 .withDriverName(driver)
                                 .withUrl(url)
                                 .withUsername("default")
                                 .withPassword("aaaaaa")
                                 .build()
        );
    }
}
